﻿using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Messaging;
using System.Runtime.Serialization.Formatters.Binary;
using System.Transactions;
using SimpleServiceBus.Bus;
using SimpleServiceBus.Bus.Transport;
using SimpleServiceBus.Utilities;

namespace SimpleServiceBus.Endpoints.Msmq.Transport
{
    public class MsmqTransport : BaseTransport 
    {
        private const string DIRECTPREFIX_OS = "DIRECT=OS:";
        private const string DIRECTPREFIX_TCP = "DIRECT=TCP:";
        private const string PREFIX_OS = "FormatName:" + DIRECTPREFIX_OS;
        private const string PREFIX_TCP = "FormatName:" + DIRECTPREFIX_TCP;

        protected MessageQueue _localQueue;
        protected MessageQueue _errorQueue;
        
        private string _inputQueue;


        private bool _purgeOnStartup;

        private readonly IDictionary<string, int> _failuresPerMessage = new Dictionary<string, int>();

        private bool NeedToAbort { get; set; }

        public bool ReceiveQueueIsTransactional { get; set; }
        public bool DestinationQueueIsTransactional { get; set; }
        public bool AttachSenderId { get; set; }
        public bool SkipDeserialization { get; set; }

        public bool PurgeOnStartup
        {
            get { return _purgeOnStartup; }
            set
            {
                _purgeOnStartup = value;
                if (value && _localQueue != null)
                    _localQueue.Purge();
            }
        }

        public string InputQueue
        {
            get
            {
                if (String.IsNullOrEmpty(_inputQueue))
                    return ReturnAddress;
                return _inputQueue;
            }
            set { _inputQueue = value; }
        }

        public override string ReturnAddress
        {
            get
            {
                if (_localQueue != null)
                    return GetIndependentAddressForQueue(_localQueue);
                if (Endpoint != null && Endpoint.Metadata != null)
                    return Endpoint.Metadata.ReturnAddress;

                return null;
            }
        }

        public MsmqTransport()
        {
            // Default to transaction destination queue
            DestinationQueueIsTransactional = true;

            AttachSenderId = true;
        }

        protected override void StartService()
        {
            if (IsSendOnly)
                return;

            BindToLocalQueue();
            BindToErrorQueue();

            base.StartService();
        }

        protected void BindToLocalQueue()
        {
            string queueName = GetFullPath(InputQueue);
            var inputQueue = new MessageQueue(queueName);
            SetLocalQueue(inputQueue);

            if (PurgeOnStartup)
                _localQueue.Purge();
        }

        protected void BindToErrorQueue()
        {
            if (!String.IsNullOrEmpty(ErrorAddress))
            {
                string queueName = GetFullPath(ErrorAddress);
                _errorQueue = new MessageQueue(queueName);
            }
        }

        //public void AbortHandlingCurrentMessage()
        //{
        //    NeedToAbort = true;
        //}

        [DebuggerNonUserCode]
        protected override void Receive()
        {
            try
            {
                _localQueue.Peek(TimeSpan.FromSeconds(1));
            }
            catch (MessageQueueException ex)
            {
                if (ex.MessageQueueErrorCode == MessageQueueErrorCode.IOTimeout)
                    return;
            }

            NeedToAbort = false;

            try
            {
                if (ReceiveQueueIsTransactional)
                    new TransactionWrapper().RunInTransaction(base.Receive);
                else
                    base.Receive();
            }
            catch (AbortHandlingCurrentMessageException)
            {
                //in case AbortHandlingCurrentMessage occurred
            }
        }

        protected override MessageEnvelope GetNextMessage()
        {
            Message m = null;
            MessageEnvelope envelope = null;
            try
            {
                m = _localQueue.Receive(GetTransactionTypeForReceive());
                if (m == null) return null;
               
                envelope = CreateTransportMessageFromMsmqMessage(m);
              
               lock (_failuresPerMessage)
                   _failuresPerMessage.Remove(m.Id);

            }
            catch (Exception ex)
            {
                bool shouldThrow;
                HandleMessageReceiveError(m, ex, out shouldThrow);
                
                if (shouldThrow) throw;
            }

            if (NeedToAbort)
                throw new AbortHandlingCurrentMessageException();

            return envelope;
        }

        public override int GetNumberOfPendingMessages()
        {
            return MsmqHelper.GetPendingMessageForQueue(_localQueue.FormatName);
        }

        public override void Send(MessageEnvelope envelope,string[] destinations)
        {
            foreach (string destination in destinations)
            {
                string address = GetFullPath(destination);

                using (var q = new MessageQueue(address, QueueAccessMode.Send))
                {
                    var toSend = new Message {BodyStream = new MemoryStream(envelope.Payload)};

                    //if (envelope.CorrelationID != null)
                    //    toSend.CorrelationId = envelope.CorrelationID;

                    toSend.Recoverable = envelope.Recoverable;

                    switch(envelope.Priority)
                    {
                        case MessagePriorities.AboveNormal:
                            toSend.Priority = MessagePriority.AboveNormal;
                            break;
                        case MessagePriorities.High:
                            toSend.Priority = MessagePriority.High;
                            break;
                        case MessagePriorities.Highest:
                            toSend.Priority = MessagePriority.Highest;
                            break;
                        case MessagePriorities.Low:
                            toSend.Priority = MessagePriority.Low;
                            break;
                        case MessagePriorities.Lowest:
                            toSend.Priority = MessagePriority.Lowest;
                            break;
                        case MessagePriorities.Normal:
                            toSend.Priority = MessagePriority.Normal;
                            break;
                        case MessagePriorities.VeryHigh:
                            toSend.Priority = MessagePriority.VeryHigh;
                            break;
                        case MessagePriorities.VeryLow:
                            toSend.Priority = MessagePriority.VeryLow;
                            break;
                    }

                    toSend.AttachSenderId = AttachSenderId;

                    //toSend.ResponseQueue = new MessageQueue(GetFullPath(envelope.ReturnAddress))
                    //                           {MachineName = Environment.MachineName};
                    toSend.Label = envelope.MessageID;
                    if (envelope.Body != null)
                        toSend.Label += ":" + envelope.Body.GetType().Name;

                    if (envelope.TimeToBeReceived < TimeSpan.MaxValue)
                        toSend.TimeToBeReceived = envelope.TimeToBeReceived;
                    toSend.UseDeadLetterQueue = envelope.UseDeadLetterQueue;
                    
                    if (envelope.Headers != null && envelope.Headers.Count > 0)
                    {
                        using (var stream = new MemoryStream())
                        {
                            var serializer = new BinaryFormatter();
                            serializer.Serialize(stream, envelope.Headers);
                            toSend.Extension = stream.GetBuffer();
                        }
                    }

                    q.Send(toSend, GetTransactionTypeForSend());
                }
            }
        }

        protected virtual MessageEnvelope CreateTransportMessageFromMsmqMessage(Message msmqMessage)
        {
            MessageEnvelope result = CreateEmptyMessageEnvelope();

            if (msmqMessage.Extension.Length > 0)
            {
                byte[] buffer = msmqMessage.Extension;
                var serializer = new BinaryFormatter();
                using (var stream = new MemoryStream(buffer))
                {
                    var headers = (Headers)serializer.Deserialize(stream);
                    if (headers != null)
                        ((IHeadersProvider)result).Headers = headers;
                }
            }

            result.Recoverable = msmqMessage.Recoverable;
            //result.TimeToBeReceived = msmqMessage.TimeToBeReceived;

            using (var payload = new MemoryStream((int)msmqMessage.BodyStream.Length))
            {
                StreamHelper.CopyStream(msmqMessage.BodyStream, payload);
                result.Payload = payload.GetBuffer();
            }

            return result;
        }

        protected virtual void HandleMessageReceiveError(Message message,Exception error,out bool shouldSuppressException)
        {
            //Log error

            if (ReceiveQueueIsTransactional)
            {
                lock (_failuresPerMessage)
                {
                    if (!_failuresPerMessage.ContainsKey(message.Id))
                        _failuresPerMessage[message.Id] = 1;
                    else
                        _failuresPerMessage[message.Id] = _failuresPerMessage[message.Id] + 1;

                    if (_failuresPerMessage[message.Id] == MaxRetries)
                    {
                        _failuresPerMessage.Remove(message.Id);

                        if (_errorQueue != null)
                            _errorQueue.Send(message, GetTransactionTypeForSend());

                        shouldSuppressException = true;
                        return;
                    }
                }
            }
            shouldSuppressException = false;
        }

        private void SetLocalQueue(MessageQueue q)
        {
            if (ReceiveQueueIsTransactional && !q.Transactional)
                throw new ArgumentException("Queue must be transactional (" + q.Path + ").");
            
            _localQueue = q;

            var mpf = new MessagePropertyFilter();
            mpf.SetAll();

            _localQueue.MessageReadPropertyFilter = mpf;
        }

        private MessageQueueTransactionType GetTransactionTypeForSend()
        {
            if (DestinationQueueIsTransactional)
            {
                if (Transaction.Current != null)
                    return MessageQueueTransactionType.Automatic;
                return MessageQueueTransactionType.Single;
            }
            return MessageQueueTransactionType.None;
        }

        private MessageQueueTransactionType GetTransactionTypeForReceive()
        {
            if (ReceiveQueueIsTransactional)
                return MessageQueueTransactionType.Automatic;
            return MessageQueueTransactionType.None;
        }

        public static string GetIndependentAddressForQueue(MessageQueue q)
        {
            if (q == null)
                return null;

            string[] arr = q.FormatName.Split('\\');
            string queueName = arr[arr.Length - 1];

            int directPrefixIndex = arr[0].IndexOf(DIRECTPREFIX_OS);
            if (directPrefixIndex >= 0)
            {
                return queueName + '@' + arr[0].Substring(directPrefixIndex + DIRECTPREFIX_OS.Length);
            }

            directPrefixIndex = arr[0].IndexOf(DIRECTPREFIX_TCP);
            if (directPrefixIndex >= 0)
            {
                return queueName + '@' + arr[0].Substring(directPrefixIndex + DIRECTPREFIX_TCP.Length);
            }

            try
            {
                // the pessimistic approach failed, try the optimistic approach
                arr = q.QueueName.Split('\\');
                queueName = arr[arr.Length - 1];
                return queueName + '@' + q.MachineName;
            }
            catch
            {
                throw new Exception(string.Concat("MessageQueueException: '",
                                                  DIRECTPREFIX_OS, "' is missing. ",
                                                  "FormatName='", q.FormatName, "'"));
            }
        }

        public static string GetFullPath(string value)
        {
            if (value.IsNonEmpty())
            {
                string[] arr = value.Split('@');

                string queue = arr[0];
                string machine = Environment.MachineName;

                if (arr.Length == 2)
                    if (arr[1] != "." && arr[1].ToLower() != "localhost")
                        machine = arr[1];

                System.Net.IPAddress ipAddress;
                if(System.Net.IPAddress.TryParse(machine, out ipAddress))
                    return PREFIX_TCP + machine + "\\private$\\" + queue;

                return PREFIX_OS + machine + "\\private$\\" + queue;
            }
            return null;
        }
        
    }
}
